chore(tracing): support for open telemetry traces - #4855

diff --git a/pkg/phlaredb/filter_profiles_bidi.go b/pkg/phlaredb/filter_profiles_bidi.go
--- a/pkg/phlaredb/filter_profiles_bidi.go
+++ b/pkg/phlaredb/filter_profiles_bidi.go
@@ -8,6 +8,7 @@
 	"github.com/grafana/dskit/tracing"
 	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
+	"go.opentelemetry.io/otel/attribute"
 	oteltrace "go.opentelemetry.io/otel/trace"
 
 	ingestv1 "github.com/grafana/pyroscope/api/gen/proto/go/ingester/v1"
@@ -82,8 +83,10 @@
 		Profile: maxBlockProfile,
 		Index:   0,
 	}, true, its...), batchProfileSize, func(ctx context.Context, batch []ProfileWithIndex) error {
-		sp.SetTag("batch_len", len(batch))
-		sp.SetTag("batch_requested_size", batchProfileSize)
+		otelSpan.AddEvent("processing batch", oteltrace.WithAttributes(
+			attribute.Int("batch_len", len(batch)),
+			attribute.Int("batch_requested_size", batchProfileSize),
+		))
 
 		seriesByFP := map[model.Fingerprint]int{}
 		selectProfileResult.Profiles = selectProfileResult.Profiles[:0]

diff --git a/pkg/phlaredb/symdb/block_reader_parquet.go b/pkg/phlaredb/symdb/block_reader_parquet.go
--- a/pkg/phlaredb/symdb/block_reader_parquet.go
+++ b/pkg/phlaredb/symdb/block_reader_parquet.go
@@ -10,6 +10,8 @@
 	"github.com/grafana/dskit/multierror"
 	"github.com/grafana/dskit/tracing"
 	"github.com/parquet-go/parquet-go"
+	"go.opentelemetry.io/otel/attribute"
+	oteltrace "go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/grafana/pyroscope/pkg/objstore"
@@ -40,10 +42,11 @@
 )
 
 func (t *parquetTable[M, P]) fetch(ctx context.Context) (err error) {
-	span, _ := tracing.StartSpanFromContext(ctx, "parquetTable.fetch")
+	span, spanCtx := tracing.StartSpanFromContext(ctx, "parquetTable.fetch")
 	span.SetTag("table_name", t.persister.Name())
 	span.SetTag("row_groups", len(t.headers))
 	defer span.Finish()
+	otelSpan := oteltrace.SpanFromContext(spanCtx)
 	return t.r.Inc(func() error {
 		var s uint32
 		for _, h := range t.headers {
@@ -55,9 +58,11 @@
 		// TODO(kolesnikovae): Row groups could be fetched in parallel.
 		rgs := t.file.RowGroups()
 		for _, h := range t.headers {
-			span.SetTag("row_group", h.RowGroup)
-			span.SetTag("index_row", h.Index)
-			span.SetTag("rows", h.Rows)
+			otelSpan.AddEvent("row_group_fetch", oteltrace.WithAttributes(
+				attribute.Int("row_group", int(h.RowGroup)),
+				attribute.Int("index_row", int(h.Index)),
+				attribute.Int("rows", int(h.Rows)),
+			))
 			rg := rgs[h.RowGroup]
 			rows := rg.Rows()
 			if err := rows.SeekToRow(int64(h.Index)); err != nil {

diff --git a/pkg/compactor/bucket_compactor.go b/pkg/compactor/bucket_compactor.go
--- a/pkg/compactor/bucket_compactor.go
+++ b/pkg/compactor/bucket_compactor.go
@@ -927,8 +927,8 @@
 			case jobChan <- g:
 			case <-maxCompactionTimeChan:
 				maxCompactionTimeReached = true
-				level.Info(c.logger).Log("msg", "max compaction time reached, no more compactions will be started")
-				sp.SetTag("msg", "max compaction time reached, no more compactions will be started")
+			level.Info(c.logger).Log("msg", "max compaction time reached, no more compactions will be started")
+			trace.SpanFromContext(workCtx).AddEvent("max compaction time reached, no more compactions will be started")
 				break jobLoop
 			}
 		}

diff --git a/pkg/frontend/readpath/queryfrontend/symbolizer.go b/pkg/frontend/readpath/queryfrontend/symbolizer.go
--- a/pkg/frontend/readpath/queryfrontend/symbolizer.go
+++ b/pkg/frontend/readpath/queryfrontend/symbolizer.go
@@ -119,6 +119,7 @@
 // profiles exist.
 func (q *QueryFrontend) shouldSymbolize(ctx context.Context, tenants []string, blocks []*metastorev1.BlockMeta) bool {
 	otelSpan := oteltrace.SpanFromContext(ctx)
+	otelSpan.AddEvent("shouldSymbolize")
 
 	if q.symbolizer == nil {
 		return false

diff --git a/pkg/ingester/pyroscope/ingest_handler.go b/pkg/ingester/pyroscope/ingest_handler.go
--- a/pkg/ingester/pyroscope/ingest_handler.go
+++ b/pkg/ingester/pyroscope/ingest_handler.go
@@ -61,7 +61,7 @@
 		msg := "failed to parse request metadata"
 		sp.LogError(err)
 		sp.SetError()
-		sp.SetTag("msg", msg)
+		trace.SpanFromContext(ctx).AddEvent(msg)
 		_ = h.log.Log("msg", msg, "err", err, "orgID", tenantID)
 		httputil.ErrorWithStatus(w, err, http.StatusBadRequest)
 		return
@@ -83,7 +83,7 @@
 		msg := "failed to read request body"
 		sp.LogError(err)
 		sp.SetError()
-		sp.SetTag("msg", msg)
+		trace.SpanFromContext(ctx).AddEvent(msg)
 		_ = h.log.Log("msg", msg, "err", err, "orgID", tenantID)
 		httputil.ErrorWithStatus(w, err, status)
 		return
@@ -95,14 +95,14 @@
 			msg := "failed to convert profile"
 			sp.LogError(err)
 			sp.SetError()
-			sp.SetTag("msg", msg)
+			trace.SpanFromContext(ctx).AddEvent(msg)
 			_ = h.log.Log("msg", msg, "err", err, "orgID", tenantID)
 			httputil.Error(w, err)
 		} else {
 			msg := "failed to ingest profile"
 			sp.LogError(err)
 			sp.SetError()
-			sp.SetTag("msg", msg)
+			trace.SpanFromContext(ctx).AddEvent(msg)
 			httputil.ErrorWithStatus(w, err, http.StatusUnprocessableEntity)
 		}
 	}

diff --git a/pkg/compactor/bucket_compactor.go b/pkg/compactor/bucket_compactor.go
--- a/pkg/compactor/bucket_compactor.go
+++ b/pkg/compactor/bucket_compactor.go
@@ -875,7 +875,7 @@
 			sp.SetError()
 			return errors.Wrap(err, "build compaction jobs")
 		}
-		sp.SetTag("discovered_jobs", len(jobs))
+		discoveredJobs := len(jobs)
 
 		// There is another check just before we start processing the job, but we can avoid sending it
 		// to the goroutine in the first place.
@@ -883,7 +883,7 @@
 		if err != nil {
 			return err
 		}
-		sp.SetTag("own_jobs", len(jobs))
+		ownJobs := len(jobs)
 
 		// Record the difference between now and the max time for a block being compacted. This
 		// is used to detect compactors not being able to keep up with the rate of blocks being
@@ -895,8 +895,17 @@
 
 		// Skip jobs for which the wait period hasn't been honored yet.
 		jobs = c.filterJobsByWaitPeriod(ctx, jobs)
-		sp.SetTag("filtered_jobs", len(jobs))
+		filteredJobs := len(jobs)
 
+		// Use AddEvent instead of SetTag to preserve per-iteration data as separate
+		// timestamped log entries, rather than overwriting previous iteration values.
+		trace.SpanFromContext(ctx).AddEvent("compaction iteration jobs",
+			trace.WithAttributes(
+				attribute.Int("discovered_jobs", discoveredJobs),
+				attribute.Int("own_jobs", ownJobs),
+				attribute.Int("filtered_jobs", filteredJobs),
+			))
+
 		// Sort jobs based on the configured ordering algorithm.
 		jobs = c.sortJobs(jobs)